Conversation
…onstruction `map_from_arrays` and `map_from_entries` built their maps without the entry checks Spark's `ArrayBasedMapBuilder` performs, so a `NULL` key inside the keys array produced a map with a `NULL` key instead of raising `NULL_MAP_KEY`, and `spark.sql.mapKeyDedupPolicy=LAST_WIN` fell the whole expression back to Spark. DataFusion 55 added `datafusion.spark.map_key_dedup_policy` and taught the `datafusion-spark` map kernels to follow it, which is the missing half. Forward Spark's `spark.sql.mapKeyDedupPolicy` to it across JNI, and pass the session's `ConfigOptions` into `ScalarFunctionExpr` so a kernel that reads a setting sees the session's value rather than DataFusion's defaults. New `SparkMapFromArrays` / `SparkMapFromEntries` / `SparkStrToMap` wrappers add the checks the upstream kernels do not perform and restate their errors as the Spark error classes `SparkErrorConverter` turns back into `QueryExecutionErrors`: a `NULL` key raises `NULL_MAP_KEY` ahead of any duplicate-key check, key and value arrays of different lengths raise `MAP_KEY_VALUE_DIFF_SIZES`, and a duplicate key under `EXCEPTION` raises `DUPLICATED_MAP_KEY` naming the key. `CometMapFromArrays` now emits `map_from_arrays`, which is null intolerant like Spark's, so the `CaseWhen` guard against NULL input arrays is no longer needed. A floating-point map key stays a documented difference: Spark normalizes `-0.0` to `+0.0` and canonicalizes `NaN` before storing a key, while the native builders compare the raw Arrow values. `spark.comet.exec.strictFloatingPoint` declines those key types. Closes apache#4680
github-actions
Bot
added
bug
peterxcli
changed the title
…d-dedup-policy # Conflicts: # native/spark-expr/src/comet_scalar_funcs.rs # native/spark-expr/src/lib.rs # native/spark-expr/src/map_funcs/mod.rs
| self.inner | ||
| .invoke_with_args(args) | ||
| .map_err(|error| as_spark_error(error, DuplicateKeyFormat::Bare)) |
There was a problem hiding this comment.
This can return a key from a preceding row. With keys [[10], [20]] and values [[100], [200]], slicing both to the second row returns {10: 200} instead of {20: 200}. The previous MapFunc returns {20: 200}.
The helper applies a zero-based keys_mask to the unsliced flat_keys, while value indices include the starting offset. I reproduced this through a native GlobalLimitExec -> ProjectionExec component test on DataFusion 55.0.0; the relevant kernels are unchanged in 55.1.0.
Please fix the offset handling in the helper or normalize the inputs before delegation, and add a sliced-list regression test. The newly enabled LAST_WIN path for map_from_entries is affected too.
| if let Some(nulls) = &key_nulls { | ||
| if nulls.slice(start, end - start).null_count() > 0 { | ||
| return Err(SparkError::NullMapKey.into()); | ||
| } |
There was a problem hiding this comment.
For keys [1, 1, NULL] under EXCEPTION, Spark 4.1.3 reports DUPLICATED_MAP_KEY, but this pre-scan reports NULL_MAP_KEY. Spark inserts entries in order and fails on the second key before reaching the null.
Please preserve that check order in both builders and update the comments claiming null-key errors always take precedence.
There was a problem hiding this comment.
Verified at 3085702: duplicate/NULL error ordering is fixed. Please also update the two map-constructor sections in map_funcs.md that still say "ahead of any duplicate-key check".
andygrove
left a comment
andygrove
left a comment
There was a problem hiding this comment.
MapBuilderSupport.keySupport only looks at the floating point gate, but MapKeySupport.keySupport a few lines above declines a non-default string collation for map lookups. Spark's ArrayBasedMapBuilder picks a collation-aware TreeMap for any StringType that is not supportsBinaryEquality, so under UTF8_LCASE the keys 'a' and 'A' are the same key, while the native builder compares raw Arrow values.
SELECT map_from_arrays(array(CAST('a' AS STRING COLLATE UTF8_LCASE), CAST('A' AS STRING COLLATE UTF8_LCASE)), array(i, i)) FROM t looks like it reaches the native path, since both casts have Literal children so CometCast folds them and supportedDataType accepts a collated StringType. Spark 4 raises DUPLICATED_MAP_KEY there and Comet returns a two-entry map. The LAST_WIN side worries me more, because the old isLastWin branch declined and sent that case back to Spark, so it used to be correct and is not after this change. Would it make sense for MapBuilderSupport.keySupport to call hasNonDefaultStringCollation the way MapKeySupport.keySupport does, with a fixture next to element_at_map_collation.sql to pin it?
The floating point note also reads as though the only difference is duplicate detection, and I think it is off in both directions. Spark only normalizes map keys from 4.0.0 onwards. spark.sql.legacy.disableMapKeyNormalization is marked .version("4.0.0") and the 3.5 ArrayBasedMapBuilder has no keyNormalizer at all, so on 3.4 and 3.5 the native builder already matches and spark.comet.exec.strictFloatingPoint declines for nothing. On 4.0 and later MapFromArrays calls mapBuilder.from(...), which reuses the original key array whenever the row has no duplicates, so a lone -0.0 key stays -0.0 in Spark too and "a -0.0 key is stored as +0.0" is not observable through map_from_arrays. MapFromEntries is the one that stores the normalized key, because it goes through put and build(), so SELECT map_from_entries(array(struct(-0.0D AS key, 1 AS value))) gives {0.0 -> 1} on Spark 4 and {-0.0 -> 1} on Comet with no duplicate anywhere. The NaN half overstates it too, since ScalarValue compares floats by to_bits, so two double('NaN') keys do collapse natively. Could the note be reworded around what actually differs per function and scoped to Spark 4.0 and later? A fixture for a floating point map key would help, since neither the default path nor the new strict decline is exercised today.
One smaller thing. The comment on checkSparkErrorParity says the mismatched-length case goes through a _LEGACY_ERROR_TEMP_* condition whose number moves between versions, but reading it out of the shipped jars it is _LEGACY_ERROR_TEMP_2128 on 3.4.3, 3.5.8, 4.0.1, 4.1.3 and 4.2.0 alike. If that holds then checkSparkError(df, "_LEGACY_ERROR_TEMP_2128") and expect_error(_LEGACY_ERROR_TEMP_2128) pin it directly and the new helper is not needed. Is there a version where the number actually differs?
On sequencing, #5846 also rewrites CometMapFromArrays.convert and adds its own SparkMapFromArrays re-export, and #5844 adds the CodegenDispatchFallback for LAST_WIN that this PR would make unnecessary. I have commented on both pointing here. Worth agreeing an order with @sunchao and @LinSimon-901101 so the same lines are not landed twice.
The upstream `datafusion-spark` map kernels read each row's entries at its own
offset but build the mask selecting the surviving keys from zero, then apply
that mask to the list's whole values array. Arrow's `filter` accepts a predicate
shorter than the array it filters, so on a sliced argument the mismatch silently
returns keys belonging to earlier rows rather than raising: keys `[[10], [20]]`
and values `[[100], [200]]`, both sliced to the second row, built `{10: 200}`
instead of `{20: 200}`. A `LIMIT` above a projection produces such an argument.
Compact any list argument whose values hold more than its offsets address
before validating or delegating, so the kernels see the layout they assume.
`map_from_entries` reached the same helper before this branch, so the bug is not
new to `map_from_arrays`; a fix belongs upstream as well.
Reported by @rich7420.
Spark's `ArrayBasedMapBuilder` inserts entries one at a time, so for keys `[1, 1, NULL]` under `EXCEPTION` it raises `DUPLICATED_MAP_KEY` on the second entry and never reaches the null. The validation pre-scanned a whole row for null keys before delegating, so it reported `NULL_MAP_KEY` instead, and its comments described the precedence as categorical rather than positional. Walk each row's keys in insertion order and raise on the first offending entry, so the two errors order the way Spark orders them, across rows as well as within one. The walk runs only when the keys carry a `NULL`: without one the kernel's own duplicate check already names the key Spark would. Under `LAST_WIN` a duplicate overwrites rather than raising, so only the null check applies. Reported by @rich7420.
`ArrayBasedMapBuilder` keys its dedup map on `TypeUtils.getInterpretedOrdering` once the key type contains a string, so under `UTF8_LCASE` the keys 'a' and 'A' are one key. The native builders compare the raw Arrow bytes and would keep both, missing the duplicate Spark reports or the overwrite Spark performs under `LAST_WIN`. `MapKeySupport` already declines a collated key for `map_extract` for the same reason; `MapBuilderSupport` only gated floating-point keys. Report `Incompatible` for a collated key type in both constructors. `CometMapFromArrays` falls back to Spark, while `CometMapFromEntries` mixes in `CodegenDispatchFallback` and stays in the Comet pipeline running Spark's own generated code. The new fixture pins both routes. Reported by @andygrove.
The note claimed Spark normalizes a floating-point map key before storing it, full stop. Two corrections, both checked against Spark's sources: `ArrayBasedMapBuilder` gained `keyNormalizer` in 4.0, alongside `spark.sql.legacy.disableMapKeyNormalization`. The 3.5 builder has no normalizer and no reference to `NormalizeFloatingNumbers`, so on 3.4 and 3.5 the native builders already match Spark and there is nothing to warn about. On 4.0+ the two functions differ. `MapFromArrays` calls `ArrayBasedMapBuilder.from`, which returns the input arrays untouched when no key repeated, so a lone `-0.0` key stays `-0.0` in Spark as it does natively; only duplicate detection diverges. `MapFromEntries` puts entries one at a time and always calls `build()`, so Spark stores the normalized key and returns `+0.0` where Comet returns `-0.0`. The gate stays unconditional. Declining on 3.4 and 3.5 costs only a fallback that `spark.comet.exec.strictFloatingPoint` users opted into. Reported by @andygrove.
…gines The length mismatch test avoided naming Spark's condition because I assumed the `_LEGACY_ERROR_TEMP_*` number moved between Spark versions, and added `checkSparkErrorParity` to `CometTestBase` to work around it. The assumption was never checked and is wrong: `mapDataKeyArrayLengthDiffersFromValueArrayLengthError` raises `_LEGACY_ERROR_TEMP_2128` in 3.4.3, 3.5.8 and 4.1.3 alike. Name the condition in the test and drop the helper, which leaves `CometTestBase` untouched by this branch. Reported by @andygrove.
6861d8b to
0021e46
Compare
|
@andygrove @rich7420 addressed review! ptal again, thanks! |
…tures `routing_map_legacy_disabled.sql` and `routing_map_legacy_enabled.sql` arrived with apache#5918 and pin how `map_from_entries` routes under `spark.sql.mapKeyDedupPolicy=LAST_WIN`. They encode the behavior this branch removes: `MapFromEntries` reported `Incompatible` under `LAST_WIN`, so `spark.comet.exec.scalaUDF.codegen.enabled` decided whether it fell back to Spark or ran through the JVM codegen dispatcher. The native builder now reads the policy from `datafusion.spark.map_key_dedup_policy`, so the expression is `Compatible` and stays native under either setting of that flag. Expect native in both fixtures. No routing coverage is lost. `map_from_entries` is still `Incompatible` for a `BinaryType` key or value, and `routing_maps_disabled.sql` and `routing_maps_enabled.sql` exercise its fallback and dispatch routes that way. `str_to_map` keeps its expectations in both fixtures: it declines for `spark.sql.legacy.truncateForEmptyRegexSplit`, which this branch does not touch.
| (builder, binaryExpr) => builder.setAnd(binaryExpr)) | ||
| // Native `map_from_arrays` is null intolerant like Spark's: a NULL keys or values array | ||
| // yields a NULL map for that row, so no CaseWhen guard is needed here. | ||
| scalarFunctionExprToProto("map_from_arrays", keysExpr, valuesExpr) |
There was a problem hiding this comment.
Please preserve short-circuit evaluation. With ANSI enabled and a Parquet row (k = NULL, v = 'bad'):
SELECT map_from_arrays(k, array(CAST(v AS INT))) FROM t;Spark returns NULL; Comet raises CAST_INVALID_INPUT because the cast runs before the NULL check. Reproduced on Spark 4.1.3; restoring the CASE guard makes it pass. Please add this regression test.
…d-dedup-policy # Conflicts: # native/spark-expr/src/comet_scalar_funcs.rs
…rcuit Spark's `BinaryExpression.eval` returns NULL as soon as the left input is NULL and never evaluates the right one, so a failing cast in the values argument does not run for a row whose keys array is NULL. Comet evaluates both argument subtrees, so removing the `CaseWhen` guard made `map_from_arrays(k, array(CAST(v AS INT)))` raise `CAST_INVALID_INPUT` under ANSI where Spark returns NULL. Restore the guard. Its `AND` over `IsNotNull` on both arguments lets the native side skip the values expression once the keys array is known NULL, which matches Spark's evaluation order. The earlier commit removed it on the grounds that native `map_from_arrays` is null intolerant; that is true of the result but says nothing about which subtrees get evaluated. Add a regression test that fails with `CAST_INVALID_INPUT` without the guard and passes with it.
|
Triage note: #5844 also makes I think the native route is the one we want here, provided DataFusion's kernel matches |
…the values expression The previous commit restored `CASE WHEN keys IS NOT NULL AND values IS NOT NULL` around `map_from_arrays`, and its regression test passed only because the table held a single row. DataFusion's `AND` skips its right side when the left side is false on every row of the batch, or on at most a fifth of them; in a batch where most rows do have keys it evaluates `values IS NOT NULL` on the whole batch, so a failing cast in the values array still runs for the NULL-keys row and raises under ANSI where Spark returns NULL. Nest one `CaseWhen` per argument instead, as apache#5846 does. DataFusion evaluates a THEN branch only on the rows its WHEN selected, so the values expression is never evaluated for a row whose keys array is NULL, which is what `BinaryExpression.eval` does. Rewrite the regression test as a five-row table written in one partition so every row shares a batch and most rows have keys. It fails with `CAST_INVALID_INPUT` against the `AND` guard and passes with the nested guards; the three `map_from_arrays` tests from apache#5846 pass as well.
`ArrayBasedMapBuilder` fixes a key's slot at its first occurrence and
only replaces its value, so `['a', 'b', 'a']` with `[1, 2, 3]` becomes
`{a -> 3, b -> 2}` rather than `{b -> 2, a -> 3}`. The datafusion-spark
kernels mirror that, but no test here told the two apart: the existing
LAST_WIN cases repeat one key, or repeat a key only in adjacent slots.
Add Rust unit tests for `map_from_arrays`, `map_from_entries` and
`str_to_map` that assert the keys and values in order, and fixture rows
for the same case. A map compares equal in any entry order, so the
fixtures pin the order through `map_keys` and `map_values`. Also cover a
NULL as the value that wins.
|
@andygrove I'd land this PR first, then #5867 rebased onto it, and close #5844 and #5846 as superseded. Ordering. datafusion-spark 55.1's #5844. This branch resolves #5589 directly: #5846. Both of its fixes are carried here. The datafusion-spark kernel validates lengths per row, and the wrapper raises #5867. Still needed: the nested guards serialize each child twice, so the |
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Reviewed 101b2da0a9272bacd4fe0bf6a719740423002be0 against base 8c229a703ccb024a8b5b1b849a56ceef0b66a4bd. The current changes address the earlier slice-offset, duplicate/null ordering, collation, and ANSI short-circuit findings. One P2 wrong-result issue remains: removing the LAST_WIN fallback admits nondeterministic children into the duplicated null guards. The inline comment has the reproduction and the connection to #5867.
CI is green: 23 successful checks and 14 skipped, including successful Required Checks. The CI run passed 1,512 native tests and 1,504 Spark 4.1 expression tests. It ran on merge revision 17ff21057639cd614dd95118c97c07403518fe10; the map wrapper, map serde, and map Scala suite match the reviewed head.
Local validation: the 21 existing map-wrapper tests passed, and focused native component probes reproduced the wrong result against a pure Spark 4.1.3 reference run over one Parquet partition. The component harness includes the current Comet map wrapper, IfExpr, and MonotonicallyIncreasingId. It uses cached DataFusion 55.0.0 after verifying the relevant map kernels and physical-expression files are byte-identical to the 55.1.0 tag. The full locked build was blocked because the configured dependency mirror does not provide DataFusion 55.1.0, so this is component validation plus a Spark reference run, not a full local Comet JNI run.
| } | ||
| } | ||
| override def getSupportLevel(expr: MapFromArrays): SupportLevel = | ||
| MapBuilderSupport.keySupport(expr.dataType.keyType) |
There was a problem hiding this comment.
[P2] Preserve fallback for nondeterministic children before enabling LAST_WIN
Could we bring in the nondeterministic-child gate from #5867 before removing the LAST_WIN decline? As noted in the sequencing discussion, the null guards still serialize each child separately from the map constructor. With default compatibility settings, this now exposes a case that previously fell back to Spark:
SET spark.sql.mapKeyDedupPolicy=LAST_WIN;
SELECT id,
map_from_arrays(
IF(monotonically_increasing_id() % 2 = 0, array(1), NULL),
array(2))
FROM t;With ids 0 through 15 in one Parquet partition, Spark 4.1.3 returns eight maps and eight nulls. A native component reproduction using the current Comet expressions and this serde's nested CASE shape returns maps only at ids 0, 4, 8, 12, turning four expected maps into NULL. The guard's counter consumes all 16 rows, while the constructor's independent counter sees only the eight rows selected by the guard.
The duplicated-child problem already exists under EXCEPTION, but removing the default LAST_WIN fallback introduces it for that policy here. #5867 is still open. Please retain fallback for nondeterministic children, incorporate its gate, or evaluate each child once before enabling this route, with a regression test for the query above. The component dependency/validation boundary is recorded in the review summary.
There was a problem hiding this comment.
Verified at 8c9ebdf43dd4b0d3f768a9101ad3994072d4e244: the new NullGuardSupport.nondeterministicChild gate fixes this finding. The focused 32-case serde probe confirms that nondeterministic children decline under both policies, including with incompatible-expression opt-in enabled, while deterministic controls remain admitted. The current CI run also passes the nondeterministic-child SQL fixture and the Scala LAST_WIN fallback regression.
…aches the null guards
The nested null guards serialize each child a second time inside the
`map_from_arrays` call, so a stateful child advances independently in
each copy: the guard's copy sees every row while the constructor's copy
sees only the rows the guard selected. With
map_from_arrays(IF(monotonically_increasing_id() % 2 != 0, array(1), NULL), array(2))
over sixteen rows in one partition, Spark returns eight maps and Comet
returned four (apache#5781). Under LAST_WIN this case used to fall back for
the policy alone, so running the policy natively exposed it there.
Port `NullGuardSupport` from apache#5867 unchanged in name, reason and
position, so that PR rebases by dropping the hunk, and decline a
nondeterministic child in `CometMapFromArrays.getSupportLevel` as
`Unsupported`; the projection falls back to Spark, which evaluates the
child once. apache#5867 still routes the same decline through the JVM codegen
dispatcher and applies it to `size`, `array_append` and `arrays_zip`.
Cover it with the query above as a Scala test on a one-partition table
under LAST_WIN, the same query in `map_from_arrays_dedup_policy.sql`,
and `map_from_arrays_nondeterministic_child.sql` for the default policy,
which mirrors the fixture in apache#5867 with `expect_fallback` in place of
`expect_dispatch`.
Fine with either order as long as the nondeterministic-child gate lands with the |
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Re-reviewed 8c9ebdf43dd4b0d3f768a9101ad3994072d4e244 against base 8c229a703ccb024a8b5b1b849a56ceef0b66a4bd. The nondeterministic-child gate fixes my previous finding. I found one additional P2: reusing an executed Dataset after changing spark.sql.mapKeyDedupPolicy can change Comet's map behavior while Spark retains its original builder policy. The inline comment includes the reproduction and mechanism. I would address this before landing; the proposed sequence of this PR followed by rebased #5867 still makes sense.
CI remains green: 23 successful checks and 14 skipped, including successful Required Checks. Run 35332676188 passed 1,547 native tests, four doc tests, and 1,506 Spark 4.1 expression tests. It ran on merge revision b6946b88d194b243eb8d5b1404a212065786c056; the relevant map implementation, serde, and regression-test sources match the reviewed head.
Local validation: the 32-case serde admission probe passed. The native map component suite passed 26 tests, including the existing 21 wrapper tests and added slice/null/policy-switch cases; a separate generated-input harness passed 4,006 cases. For the new finding, a Spark 4.1.3 Parquet probe verified the same QueryExecution and executedPlan across two actions: all three Spark map constructors retained their LAST_WIN results, while a diagnostic UDF calling the exact current Comet configuration serializer observed EXCEPTION in the second action. Separate exact-wrapper assertions confirmed that the same inputs then raise DUPLICATED_MAP_KEY under EXCEPTION. Base/current serde probes also confirmed that this LAST_WIN map_from_arrays expression previously fell back and is now admitted natively.
Validation limit: these are separate Spark, serializer, and native component executions, not a full current-head Comet/JNI reproduction. Native probes use cached DataFusion 55.0.0 / Arrow 59.3.0 with a small SparkError display stub; the relevant 55.1.0 map sources were inspected and compared. The locked local build remains blocked by dependency availability and the configured package-security proxy.
| builder.putEntries( | ||
| SQLConf.MAP_KEY_DEDUP_POLICY.key, | ||
| SQLConf.get.getConf(SQLConf.MAP_KEY_DEDUP_POLICY).toString) |
There was a problem hiding this comment.
[P2] Preserve the map policy across repeated actions
This reads the current task SQLConf on every new native iterator, while Spark's ArrayBasedMapBuilder captures the policy once and the executed expression retains that builder. Reusing the same Dataset after changing the policy therefore gives different behavior:
val path = java.nio.file.Files.createTempDirectory("map-policy").resolve("data").toString
spark.range(0, 1, 1, 1).write.parquet(path)
spark.conf.set("spark.sql.mapKeyDedupPolicy", "LAST_WIN")
val df = spark.read.parquet(path)
.selectExpr("map_from_arrays(array(id, id), array(1, 2)) AS m")
df.collect() // {0 -> 2}
spark.conf.set("spark.sql.mapKeyDedupPolicy", "EXCEPTION")
df.collect() // Spark still returns {0 -> 2}In the second action, the exact current Comet serializer sends EXCEPTION; CometExecRDD creates a fresh iterator/native session, and the native wrapper raises DUPLICATED_MAP_KEY for these inputs. The old LAST_WIN fallback preserved Spark's behavior. The same policy-capture mismatch affects map_from_entries and str_to_map.
Please preserve the policy with the expression/plan and add a regression that executes the same Dataset twice across a policy change. Constructing a new Dataset for each policy does not cover this case. I executed the Spark reference, exact serializer, and native wrapper effects separately; the full JNI validation limitation is recorded in the review summary.
… native iterator Spark's `ArrayBasedMapBuilder` reads `spark.sql.mapKeyDedupPolicy` when an expression is first evaluated and keeps that builder, so a Dataset executed again after the session setting changed still builds its maps under the policy it started with. `CometExecIterator` forwarded the setting from the task SQLConf on every new native iterator instead, so the same Dataset switched behavior: executed under LAST_WIN, it raised DUPLICATED_MAP_KEY on its next action once EXCEPTION was set, where Spark kept returning the LAST_WIN map. Carry the policy with the expression. New `MapFromArrays`, `MapFromEntries` and `StrToMap` proto messages hold a `map_key_dedup_policy` that the serde reads when it converts the plan, which the Dataset reuses across actions, the way `Hour` carries its timezone. The planner builds the native wrappers with that policy, and they hand it to the datafusion-spark kernels through the session options those kernels read, so whatever the session holds at execution time does not apply. Drop the per-iterator forwarding and the session-level `datafusion.spark.map_key_dedup_policy` setting, and the name-based registrations the dedicated messages replace. Cover it with a test that executes one Dataset twice across a policy change, for all three constructors and in both directions, with Comet disabled and enabled; it failed on the second `collect()` before this change. A native unit test checks that the wrapper's policy wins over the session option.
|
Re-reviewed [P2] Align policy capture with Spark's builder initialization
spark.conf.set("spark.sql.adaptive.enabled", "false")
val path = java.nio.file.Files.createTempDirectory("map-policy").resolve("data").toString
spark.range(0, 1, 1, 1).write.parquet(path)
spark.conf.set("spark.sql.mapKeyDedupPolicy", "EXCEPTION")
val df = spark.read.parquet(path)
.selectExpr("map_from_arrays(array(id, id), array(1, 2)) AS m")
df.explain()
spark.conf.set("spark.sql.mapKeyDedupPolicy", "LAST_WIN")
df.collect()Spark returns Could we align capture with Spark's builder initialization while keeping the policy stable on subsequent actions, and add coverage for explaining before first execution? The new repeated-Dataset test executes the query before changing the setting, so it does not cover this case. Validation: I reproduced all six cases end to end with the exact-head JNI library, the unmodified DataFusion 55.1.0 lockfile, Spark 4.1.3, and JDK 17. The probes cover all three constructors in both directions, assert that a Comet projection is present before execution, and verify that the explained plan is reused. All 35 existing map-suite tests and 22 native wrapper tests passed. The six diagnostic probes pass when the mismatch is reproduced; the root-reactor Maven run and its XML reports confirm all 41 selected JVM tests completed. The native build disabled the optional HDFS feature, and these probes use local Parquet. CI: 23 checks succeeded and 14 were skipped, including successful Required Checks. Run 35435933319 passed 1,599 native tests, four separate allocator-accounting tests, and 1,530 Spark expression tests. CI ran merge revision |
…p it Spark's `ArrayBasedMapBuilder` is a lazy field of the map expression, so `spark.sql.mapKeyDedupPolicy` is read the first time the expression is evaluated and the expression keeps that builder afterwards. The previous commit read the setting in the serde, which runs when the plan is converted, and `explain()` converts a plan without evaluating anything. A Dataset explained under EXCEPTION and then first collected under LAST_WIN therefore raised DUPLICATED_MAP_KEY where Spark returns the map, and the reverse direction returned a map where Spark raises. Capture the policy in a `lazy val` on `CometNativeExec` instead. It is forced by the first `doExecuteColumnar`, which `explain` does not reach, and the node lives in the cached `executedPlan`, so every later action reuses the value. The captured value reaches native through the plan's config map, so `serializeCometSQLConfs` no longer reads the task SQLConf for this setting; the shuffle-writer and write paths pass it the same way. That makes the proto messages the previous commit added unnecessary: the problem was never where the policy travels but when it is read. Revert them, along with the planner arms, the serde overrides and the wrappers' constructor policy, so the constructors are wired as they were and the kernels read the session option again. Cover the new case with a test that materializes the plan under one policy and first executes it under the other, in both directions and with Comet disabled and enabled; it fails both ways without this change.
github-actions
Bot
added
area:writer
|
Reviewed [P2] Preserve policy changes outside whole-stage codegen — operators.scala:588 The new lazy value always retains the first policy. Spark rebuilds its map builder in each task when the projection runs outside whole-stage codegen, including wide projections exceeding the default field limit. With AQE disabled and a 101-column projection:
The reverse switch also diverges. Reproduced end to end for all three constructors in both directions, with wide projections and explicitly disabled whole-stage codegen. The previous CI: only the labeling check exists for this head. GitHub reports merge conflicts. |
…d-dedup-policy # Conflicts: # spark/src/main/scala/org/apache/comet/CometExecIterator.scala # spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometNativeShuffleWriter.scala # spark/src/main/scala/org/apache/spark/sql/comet/operators.scala
…e whole-stage codegen Spark reads `spark.sql.mapKeyDedupPolicy` into `ArrayBasedMapBuilder`, a lazy field of the map expression, so when it reads it depends on how the projection runs. Outside whole-stage codegen the projection is rebuilt in every task and the setting is read again on each action; inside it the builder is created once on the driver, in the first action, and kept. The previous commit froze the policy on the first execution, which matches the whole-stage case. Measuring both engines across the two codegen paths, the two directions of a policy change and both the repeated-action and materialize-then-execute scenarios, that freeze matched Spark in five of eight comparisons; reading the setting when each native plan is built matches seven of eight. The freeze also diverges by returning a map where Spark raises `DUPLICATED_MAP_KEY`, while reading per plan diverges by raising where Spark returns a map, so the remaining difference is loud rather than silent. Drop the freeze and the parameter it threaded through `NativeExecContext`, `CometExecRDD`, `CometExecIterator`, the shuffle writer and both write execs. The one case that cannot also be matched, a Dataset executed more than once across a change to the setting with the projection inside whole-stage codegen, is recorded in the map_funcs expression audit: Comet replaces the operator before `CollapseCodegenStages` runs, so the plan it sees carries no record of which path Spark would have taken. Cover both scenarios: one test runs a Dataset twice across a change in each direction, in both of the configurations that leave whole-stage codegen, and one materializes the plan under one policy and first executes it under the other.
The Preflight job runs `prettier --check "**/*.md"`, which normalizes emphasis to underscores.
|
Heads up that I approved #5846 and it should land ahead of this one, so this will need a rebase. It is the smaller of the two and it fixes silent wrong results, so I did not want to hold it behind the policy-capture question here. The overlap resolves cleanly in one direction. The One thing is worth carrying over rather than dropping. #5846 asserts the length mismatch in |
andygrove
left a comment
andygrove
left a comment
There was a problem hiding this comment.
The null key and dedup policy fixes look right to me now. Reading the policy when the task builds its native plan puts it where Spark's interpreted path reads it too. A few things to pick up when you rebase onto #5846.
The comment above map constructors follow a dedup policy change between actions in CometMapExpressionSuite still has the paragraph from the earlier approach, which says Comet captures the policy when it converts the plan. The paragraph under it says the opposite and matches the code. Could the first one go?
map_funcs.md still says a NULL key raises NULL_MAP_KEY ahead of any duplicate-key check, in both the map_from_arrays and map_from_entries sections, and the comments in map_from_arrays.sql and map_from_entries.sql say the same. The builders now report whichever of the two comes first in the row, which is what Spark does. Could those say that instead, as @rich7420 asked?
Nothing exercises the spark.comet.exec.strictFloatingPoint branch of MapBuilderSupport.keySupport yet, so deleting it would leave the suite green. Could we add a fixture with a DOUBLE key under -- Config: spark.comet.exec.strictFloatingPoint=true that pins where each constructor goes?
|
ON it |
comphead
left a comment
comphead
left a comment
There was a problem hiding this comment.
Reviewed at 88c877b. Line numbers below refer to that commit. Everything here comes from reading the code. Nothing was built or run.
The approach looks right to me. Wrapping the datafusion-spark kernels and reading the policy per task matches Spark outside whole-stage codegen. The items below are mostly about test strength, redundancy and per-batch cost.
Major
1. The two dedup-policy timing tests never re-run df (CometMapExpressionSuite.scala:255-333)
- Problem:
checkAnswer(df.select("a", "e", "s"), …)(lines 274, 287 and 318) creates a newQueryExecution, so every "run again" step plans a fresh query instead of re-runningdf. The.selectcame in with the padding columns in 362f288. Before that, 6962053 calledcheckAnswer(df, …). - Consequence:
- The "freeze the policy at first execution" design that 362f288 rejected would pass both tests.
- The first half of the "materialize then execute" test would pass even with plan-time capture.
- So the tests don't pin the design the PR settled on.
- Fix:
- Assert on
dfitself: compare the first three fields ofdf.collect(), or add the padding values to the expected row. - Then one sequence per config covers both rejected designs: materialize under policy A, execute under B, re-execute under A.
assertDuplicateMapKeyre-implements thestructuredErrorhelper nested insidecheckSparkError, but drops its "noCometNativeExceptionin the cause chain" check. Make that helperprotectedand reuse it.
- Assert on
Minor
2. The length pre-check changes which error Spark would report (map_builders.rs:259-293)
- Problem: When no key is NULL,
validate_map_from_arrayschecks every row's lengths before the kernel looks for duplicate keys.- Take row 0 =
[1,1]→[a,b]and row 1 =[2]→[c,d]. - Comet raises
_LEGACY_ERROR_TEMP_2128(length mismatch). - Spark raises
DUPLICATED_MAP_KEYfor row 0.
- Take row 0 =
- Why it's avoidable: the kernel already checks lengths row by row, in Spark's order (
utils.rs:199-204in datafusion-spark 55.1). - Fix: return early when no row that gets built has a NULL key, as the entries validator already does. Translate the kernel's length message in
as_spark_error, with a pin test like the one for the duplicate-key message. This also removes twoVec<usize>offset copies per batch.
3. The inner values IS NOT NULL guard no longer does anything (maps.scala:242-265)
- Why it's redundant:
- #5846 needs the inner guard because DataFusion's
MapFunconly honours NULL key rows. - This PR switches to the datafusion-spark kernel. That kernel returns NULL for a NULL values row (
utils.rs:196-199), and the wrapper skips such rows (map_builders.rs:281). - Spark's
BinaryExpression.evalevaluatesrightwheneverleftis non-null. SoCASE WHEN keys IS NOT NULL THEN map_from_arrays(keys, values) ENDkeeps Spark's evaluation order.
- #5846 needs the inner guard because DataFusion's
- What it saves:
valuesis evaluated once instead of twice, and one filter-plus-merge disappears on batches where a values array is NULL.NullGuardSupportcould then check onlyexpr.left. - Knock-on: rows with a NULL values array would then reach the wrapper. Its NULL walk should only consider rows that actually get built (see item 4).
- Separate small fix: drop
setElseExpr(null).create_case_exprwraps the ELSE in aCast, and DataFusion's NULL-ELSE shortcut only recognizes a bare nullLiteral.
4. map_from_entries takes the slow path on batches that won't fail (map_builders.rs:295-327)
- Problem: the per-entry
ScalarValuewalk starts whenever the key child has any NULL. A NULL struct entry usually has a NULL key underneath it, and that row just becomes a NULL map, not an error. - Cost: under
EXCEPTION, one such row makes every other row in the batch pay atry_from_arrayplus aHashSetinsert per entry. The kernel then repeats the same work. - Fix:
- Only enter the walk when some valid entry has a NULL key. That is true when the null count of
NullBuffer::union(Some(&key_nulls), structs.nulls())exceeds the null count ofstructs.nulls(), which costs one AND and a popcount. - Do that check before line 302 builds the offsets
Vec. - In
check_keys_in_order,seen.insert(key.clone())clones every key in case it is needed for the error message.HashSet::replacehands back the duplicate without the clone.
- Only enter the walk when some valid entry has a NULL key. That is true when the null count of
5. The sliced-list workaround has no exit and copies too much (map_builders.rs:218-257)
- No exit: there is no link to apache/datafusion#25419. It is fixed on DataFusion main by apache/datafusion#25431, which is not in 55.1.
explode.rsandplanner.rsalready carry "delete once Comet moves to a DataFusion release carrying apache/datafusion#N" notes, so there's a pattern to follow. - Copies too often:
- Only a nonzero first offset breaks the kernel's mask. Arrow's
filteraccepts a shorter predicate, so a head slice (for example fromLIMIT) is safe. - But
entries_start_at_zeroalso requires the last offset to equalvalues.len(), so head slices get deep-copied anyway.
- Only a nonzero first offset breaks the kernel's mask. Arrow's
- Copies too much:
takecopies every nested child. A zero-copy rebase (shifted offsets plusvalues.slice(first, last - first)) is what #25431 does, and it costs O(rows). - Dead code: the
FixedSizeListarm can never fire.
6. Doc errors
- Wrong error class:
map_funcs.md:53and the module doc ofmap_builders.rssay Spark raisesMAP_KEY_VALUE_DIFF_SIZES. Spark users actually see_LEGACY_ERROR_TEMP_2128, becauseShimSparkErrorConvertermaps the native error tomapDataKeyArrayLengthDiffersFromValueArrayLengthError(). The PR's own test pins_LEGACY_ERROR_TEMP_2128. - Wrong comment on
expand_scalars: it says the kernel "expands them anyway". That is false when every argument is a scalar.make_scalar_functioncomputes one row and returns aScalar, while the wrapper computesnumber_rowsrows. It rarely matters in production because Spark constant-folds all-literal calls.CometConcatWsshows the in-repo way to handle this case.
Simplification
Tests to remove or move (the second bullet applies once #5846 lands):
- Nondeterministic-child duplicates: remove the Scala test "a nondeterministic child falls back under LAST_WIN", plus the last query and its table in
map_from_arrays_dedup_policy.sql. They repeatmap_from_arrays_nondeterministic_child.sql, andgetSupportLevelnever reads the policy. - Overlap with #5846: remove the Scala tests "a null keys array skips…", "…different lengths are rejected" and "a null input array gives a null map". #5846 has the same three cases.
- LAST_WIN halves: remove the LAST_WIN halves of both "duplicate key follows…" tests. The
map_keys/map_valuescolumn queries in the fixtures already pin that behavior. - "a null entry gives a null map": turn the Scala test into a table row in
map_from_entries.sql. - Rust tests to drop:
- the five LAST_WIN value tests (
*_honours_last_win×3,*_keeps_…_under_last_win×2) *_rejects_null_key×2rejects_key_value_length_mismatch- These only exercise the upstream kernel, and the fixtures check the same behavior against Spark. Also merge the two unquoted-key message pin tests into one.
- the five LAST_WIN value tests (
- Routing fixtures: in all three
routing_map_legacy_*.sqlfiles (including_opt_in, which the PR didn't touch), remove:- the
map_from_entriesquery, its comment and theecolumn - the
LAST_WINandMapFromEntries.allowIncompatibleconfig lines - First move the "stays native" check into the
*_dedup_policy.sqlfiles withexpect_native(map_from_entries)andexpect_native(str_to_map). A plainqueryalso passes when the expression goes through the JVM dispatcher, so it doesn't prove native execution.
- the
- Weaker or duplicate fixture queries:
- literal queries that repeat a table row (keep one literal query per file for the scalar path)
- the column query that doesn't pin entry order
str_to_map(s, ',', ':'), which is the same expression asstr_to_map(s)- the unused
kandvcolumns intest_map_from_arrays_nondet
Keep the checkSparkError, slice, error-ordering and message-pin tests. expect_error checks neither native execution nor that the error was converted. For example, in str_to_map.sql, expect_error(DUPLICATED_MAP_KEY) also matches the raw upstream message. expect_error(Duplicate map key a was found) should pin the converted one, but I haven't verified that. The slice fix can be tested end to end with the suite's existing ORDER BY … LIMIT … OFFSET pattern, which I'd use through map_from_entries.
Code
- Wrapper structs: use
#[derive(Debug, Default, PartialEq, Eq, Hash)], asCometCollectListandCometConcatWsdo. The threenew()functions have no callers. element_validity: it can belogical_nulls().filter(|n| n.null_count() > 0), which also catches nulls inside dictionary values.- Unreachable code:
- The
LargeListandFixedSizeListarms can't be reached, because Comet maps every SparkArrayTypetoList. - The row-count check can't fire after
expand_scalars.
- The
- Lint allow:
clippy::allow_attributesin the allow list is DataFusion's lint policy, not Comet's.agg_funcs/mode.rsusesScalarValuehash keys with no allow at all. Worth confirming with clippy. jni_api.rs: preferoptions_mut().spark.map_key_dedup_policy = policy.parse()?overset_str, which unwraps. A future rename then fails to compile instead of panicking on every plan.planner.rs:3673:session_config_options()has one caller, yet its doc describes it as the general way to get the options.- Other sites still pass
ConfigOptions::default():planner.rslines 814, 1128, 1147, 1163 and 1193, five sites intemporal.rs, andmodulo_expr.rs:282. - Either inline it (as at
:2342), or capture the options once inPhysicalPlanner::newand use them everywhere (fine as a follow-up). - No kernel at those other sites reads config today, so this is a trap for later, not a bug.
- Other sites still pass
maps.scala: list the strict floating-point decline ingetIncompatibleReasons(), asCometSortArrayandCometMapSortdo. The note also says "fall back to Spark", butmap_from_entriesgoes through the JVM dispatcher instead.NullGuardSupport: it has one caller, and its doc describes dispatcher routing that caller doesn't have.CometElementAt(arrays.scala:651) already declines the same case inline. Either inline the object until #5867 adds more callers, or use it fromCometElementAttoo.
Comments: several explanations are repeated many times. Policy forwarding is explained in more than a dozen places, and the nondeterministic-child and AND-guard reasoning and the floating-point note in several each. Keep one copy in the audit doc and point to it. Audit lines 51 and 63 are identical bullets. Replace the pull/5854#discussion_r… links (2 in maps.scala, 4 in the suite) with issue links. Main code has only one other link to a PR review thread.
Questions
- Floating-point keys: map lookups decline them always (
MapKeySupport), but the builders decline them only understrictFloatingPoint. Both divergences from Spark are silent. Is the difference intentional? - Key text in
DUPLICATED_MAP_KEY: the key is rendered byScalarValue's Display, not by Spark'stoStringof the internal value. ADOUBLEkey prints1instead of Spark's1.0, and aDATEkey prints2024-01-01instead of the day count. Spark's owncheckError(parameters = …)tests would catch this. Is that acceptable? - Upstream issue: would you file a datafusion-spark issue for the missing NULL-key check and for a typed duplicate-key error, and link it from a TODO? Until it's fixed, the two list builders could treat an upstream
[DUPLICATED_MAP_KEY]as a signal and re-runcheck_keys_in_orderon that rare path to find the key. That would drop the parsing of the unquoted message.
Assessment
- Scope:
- Right-sized for the fix, apart from
NullGuardSupportbeing added ahead of #5867. - The planner change passes the session options to every UDF built by
create_scalar_function_expr. Comet's session differs from the defaults only in: target partitions, batch size, parquet pushdown, the skip-partial-aggregation threshold, the dedup policy, andspark.comet.datafusion.*pass-through. - Of those, only the dedup policy is read by any scalar kernel, though a pass-through override would now reach UDFs too.
- Right-sized for the fix, apart from
- Spark compatibility: I checked
ArrayBasedMapBuilderandMapFromEntriesin Spark 3.4.3, 3.5.8 and 4.1.1.- Matches Spark: NULL keys, the dedup policy, the first-position order under LAST_WIN, and reading the policy per task outside whole-stage codegen.
- Documented gaps: re-execution inside whole-stage codegen, and floating-point keys on 4.0+.
- Undocumented gaps: item 2 and the key rendering.
- Tests:
- Apart from item 1, the new tests look like they would fail if the fix were reverted.
ci.mdrecommends therun-spark-4.1-testslabel for serde and planner changes, and it hasn't been applied.- The
area:writer,area:shuffleandarea:Iceberglabels look left over from an earlier revision.
- Performance: there's no benchmark, which is fine for a correctness fix. Items 3 to 5 add per-batch costs on the common path. I estimated those from the code and did not measure them.
Already raised by others and not repeated here: the stale test comment, the "ahead of any duplicate-key check" wording, the strictFloatingPoint fixture, and deleting #5846's map_from_arrays.rs and "map" arm on rebase.
6 participants
Which issue does this PR close?
Rationale for this change
Spark builds every map through
ArrayBasedMapBuilder, which refuses aNULLkey and resolves duplicate keys according tospark.sql.mapKeyDedupPolicy. Comet'smap_from_arraysandmap_from_entriesdid neither. ANULLinside the keys array produced a map with aNULLkey instead of an error, and setting the policy toLAST_WINpushed the whole expression back to Spark.DataFusion 55 supplies what was missing. Its
datafusion.spark.map_key_dedup_policyoption takes the sameEXCEPTIONandLAST_WINvalues as the Spark config, and thedatafusion-sparkmap kernels already follow it. Once Comet passes the setting through,LAST_WINruns natively, and the remaining checks thatArrayBasedMapBuilderperforms cost only a few lines on top.What changes are included in this PR?
spark.sql.mapKeyDedupPolicynow crosses JNI.CometExecIterator.serializeCometSQLConfssends it explicitly, sincecometSqlConfscarries only keys underspark.comet., andprepare_datafusion_session_contextapplies it to the session asdatafusion.spark.map_key_dedup_policy.A second change was needed before that setting could reach a kernel at all.
create_scalar_function_exprhanded everyScalarFunctionExpra freshConfigOptions::default(), so any kernel reading a session option saw DataFusion's defaults. It now passes the session's ownConfigOptions.native/spark-expr/src/map_funcs/map_builders.rsadds three wrappers,SparkMapFromArrays,SparkMapFromEntriesandSparkStrToMap. Each calls the matchingdatafusion-sparkkernel, adds the checks that kernel skips, and translates its errors into the Spark error classes thatSparkErrorConverterconverts back intoQueryExecutionErrors:NULLkey raisesNULL_MAP_KEY, before any check for duplicates, matching the order Spark applies them;MAP_KEY_VALUE_DIFF_SIZES;EXCEPTIONraisesDUPLICATED_MAP_KEYand names the key.str_to_mapneeds only the last of these, because splitting a string never yields aNULLkey. Passing the config through also fixed itsLAST_WINcase, which used to raise an error where Spark returns a map.CometMapFromArraysnow emitsmap_from_arrays. It used to emit the genericmapwrapped inCaseWhen(IsNotNull(left) AND IsNotNull(right), ...)so that a NULL input array yielded a NULL map; the Spark kernel already behaves that way, so the wrapper came out. Both serdes also drop theirLAST_WINIncompatiblebranch.One difference with Spark remains.
ArrayBasedMapBuildernormalizes a floating point key before storing it, so-0.0becomes+0.0and everyNaNcollapses into one. The native builders compare the raw Arrow values, so a map built from both-0.0and+0.0keeps two entries where Spark reports a duplicate key. The compatibility notes record this, andspark.comet.exec.strictFloatingPointmakes Comet decline a floating point key type for anyone who needs the guarantee.How are these changes tested?
The 21 native unit tests cover the wrappers. Two of them pin the exact wording DataFusion uses when it reports a duplicate key, because the wrapper reads that message to recover the key it should name. If DataFusion rewords the message, those tests fail rather than the error quietly degrading into a generic execution failure.
Seven new tests in
CometMapExpressionSuiterun each case through both engines and compare the exception type, error class and SQLSTATE, along with the answers each engine returns underLAST_WIN.Among the SQL fixtures, the two
*_dedup_policy.sqlfiles used to assert theLAST_WINfallback and now assert native execution.map_from_arrays.sql,map_from_entries.sqlandstr_to_map.sqlgained theEXCEPTIONerror cases, andstr_to_map_dedup_policy.sqlis new. That also retires theTODO: Add LAST_WIN policy tests when spark.sql.mapKeyDedupPolicy config is supportednote instr_to_map.sql.The test for mismatched array lengths compares the two engines against each other instead of naming an error condition. Spark still reports that case through a
_LEGACY_ERROR_TEMP_*condition whose number moves between Spark versions, soCometTestBase.checkSparkErrornow builds on a newcheckSparkErrorParityhelper.The
ConfigOptionschange affects every scalar function, so the full 487-fixture suite ran green as well.